package com.atguigu.flink.sql;

import com.atguigu.flink.beans.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * @author Felix
 * @date 2023/12/13
 * 该案例演示了流和表直接的转换
 */
public class FlinkSQL02_Stream2Table2Stream {
    public static void main(String[] args) throws Exception {
        //指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //执行表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //准备流
        DataStreamSource<WaterSensor> sensorDS = env.fromElements(
            new WaterSensor("s1", 1L, 1),
            new WaterSensor("s1", 2L, 2),
            new WaterSensor("s2", 2L, 2),
            new WaterSensor("s3", 3L, 3),
            new WaterSensor("s3", 4L, 4)
        );

        //将流转换为动态表
        Table sensorTable = tableEnv.fromDataStream(sensorDS);

        //指定只查询流中的某些列
        // Table sensorTable = tableEnv.fromDataStream(sensorDS, $("id"), $("vc"));
        // 将timestamp字段重命名为ts
        // Table sensorTable = tableEnv.fromDataStream(sensorDS, $("id").as("sid"), $("vc"));



        // tableEnv.createTemporaryView("aaa",sensorDS);
        // tableEnv.executeSql("select * from aaa").print();
        tableEnv.executeSql("select * from " + sensorTable).print();
        // sensorTable.execute().print();

        //将动态表转换为流

        tableEnv.createTemporaryView("sensor", sensorTable);

        Table filterTable = tableEnv.sqlQuery("select id,ts,vc from sensor where ts>2");
        Table sumTable = tableEnv.sqlQuery("select id,sum(vc) from sensor group by id");

        //追加流
        // DataStream<Row> filterDS = tableEnv.toDataStream(filterTable);
        DataStream<WaterSensor> filterDS = tableEnv.toDataStream(filterTable, WaterSensor.class);

        //回撤流
        DataStream<Row> sumDS = tableEnv.toChangelogStream(sumTable);

        filterDS.print("追加");
        sumDS.print("回撤");

        env.execute();
    }
}
